package operator.transfer;

import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import util.Printer;
import util.ThreadHelper;

class BufferOperator {

    public static void main(String[] args) {
//        buffer1();
//        buffer2();
//        buffer3();
        buffer4();
    }

    // count = skip
    static void buffer1(){
        Observable.range(1, 10)
                .buffer(2)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        for (Integer i: integers) {
                            Printer.print("integer = " + i);
                        }
                        Printer.print("-----------------");
                    }
                });
    }

    //从发射第一次count数据开始，按步长再次发射count
    static void buffer2(){
        Observable.range(1, 10)
                .buffer(2, 4)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        for (Integer i: integers) {
                            Printer.print("integer = " + i);
                        }
                        Printer.print("-----------------");
                    }
                });

        //应用：只取奇数
//        Observable.range(1, 10)
//                .buffer(1, 2)
//                .subscribe(new Consumer<List<Integer>>() {
//                    @Override
//                    public void accept(List<Integer> integers) throws Exception {
//                        for (Integer i: integers) {
//                            Printer.print("integer = " + i);
//                        }
//                        Printer.print("-----------------");
//                    }
//                });
    }

    //按固定时间间隔搜集发射出来的数据
    static void buffer3(){
        Observable.intervalRange(0, 10, 1, 1, TimeUnit.SECONDS)
                .buffer(600, TimeUnit.MILLISECONDS)
                .subscribe(new Consumer<List<Long>>() {
                    @Override
                    public void accept(List<Long> longs) throws Exception {
                        Printer.print(longs.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {

                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        Printer.print("buffer3 completed");
                    }
                });
        ThreadHelper.sleep();
    }

    //按固定时间间隔或者数量搜集发射出来的数据
    static void buffer4(){
        Observable.intervalRange(0, 10, 1, 1, TimeUnit.SECONDS)
                .buffer(600, TimeUnit.MILLISECONDS, 2)
                .subscribe(list -> Printer.print(list.toString()));
        ThreadHelper.sleep();
    }


}
